package ymqtt

import (
	"context"
	"encoding/json"
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"time"
)

type resultData struct {
	ClientId    string `json:"client_id"`
	RequesterId string `json:"requester_id"`
	Result      bool   `json:"Result"`
}

type RequesterData struct {
	ClientId    string      `json:"client_id"`
	RequesterId string      `json:"requester_id"`
	NeedResult  bool        `json:"need_result"`
	Data        interface{} `json:"data"`
}

type Client struct {
	*handle
	mqtt mqtt.Client
}

func (c *Client) init() {
	c.subscribeDeviceResponse()
	c.subscribeDeviceTest()
}

func (c *Client) subscribeRegister(topic string, qos byte, callback mqtt.MessageHandler) {
	token := c.mqtt.Subscribe(topic, qos, callback)
	token.Wait()
	if token.Error() != nil {
		panic(token.Error())
	}
}

func (c *Client) subscribeDeviceResponse() {
	c.subscribeRegister("service/device/response", 2, func(mq mqtt.Client, message mqtt.Message) {
		var data resultData
		payload := message.Payload()
		err := json.Unmarshal(payload, &data)
		fmt.Printf("subscribe: topic[%s] retained[%t] Payload[%s] err: %s \r\n", message.Topic(), message.Retained(), payload, err)
		if err == nil {
			c.deviceGet(data.ClientId).result.write(data.RequesterId, Msg{Time: time.Now(), Result: data.Result})
		}
	})
}

func (c *Client) subscribeDeviceTest() {
	c.subscribeRegister("service/device/test", 2, func(c mqtt.Client, message mqtt.Message) {
		fmt.Printf("topic[%s] MessageID[%d] Retained[%s] Payload[%s] \r\n", message.Topic(), message.MessageID(), message.Retained(), message.Payload())
	})
}

func (c *Client) Publish(topic string, qos byte, retained bool, request RequesterData) (r bool, e error) {
	payload, err := json.Marshal(request)
	if err != nil {
		e = err
		return
	}
	token := c.mqtt.Publish(topic, qos, retained, payload)
	token.Wait()
	e = token.Error()
	if e != nil {
		return
	}

	if request.NeedResult {
		num := 0
		ticker := time.NewTicker(time.Second)
		defer ticker.Stop()
		device := c.deviceGet(request.ClientId)
		for _ = range ticker.C {
			if num > 9 {
				fmt.Printf("requestId:%s timeout Result:false \r\n", request.RequesterId)
				break
			}

			msg := device.result.read(request.RequesterId)
			if msg != nil {
				fmt.Printf("Msg: %+v", msg)
				r = msg.Result
				break
			}
			num++
		}
	} else {
		r = true
	}

	return
}

func NewClient(ctx context.Context, conf Conf) *Client {
	opts := mqtt.NewClientOptions().AddBroker(conf.Dns).SetClientID(conf.ClientID).SetKeepAlive(time.Second * time.Duration(conf.KeepAlive)).SetPingTimeout(time.Second * time.Duration(conf.PingTimeout))
	if conf.UserName != "" {
		opts.SetUsername(conf.UserName)
	}
	if conf.Password != "" {
		opts.SetPassword(conf.Password)
	}
	fmt.Printf("%+v \r\n", opts)

	c := mqtt.NewClient(opts)
	if token := c.Connect(); token.Wait() && token.Error() != nil {
		panic(token.Error())
	}

	client := &Client{
		mqtt:   c,
		handle: newHandle(ctx),
	}
	client.init()

	return client
}
